<?php

namespace App\Service\CommodityService;

use App\Service\BaseService;
use App\Model\CommodityModel\WarehouseInfo;
use Hyperf\Amqp\Producer;
use App\Service\Code\Normal;
use App\Amqp\Producer\SupplierCommodityPullProducer;
use App\Service\AppService\SystemCacheConfigService;
use App\Model\CommodityModel\SupplierCompany;
use Hyperf\DbConnection\Db;
use Hyperf\Utils\WaitGroup;

/**
 * 供应商商品计划任务拉取中心 
 * @author chenshubo
 */
class SupplierCommodityPullLogService extends BaseService
{
    const TIME_WAIT = 60;//分钟
    const PULL_USER_COUNT = 10;//每次同时拉取供应商的数量
    
    public function runTask( $SupplierID = 0 )
    {
        //系统业务配置里面管控配置参数
        $config = SystemCacheConfigService::instance()->getCacheInfoBySystemCode('channel_task_user_limit');
        if( empty($config) ) {
            return Normal::returnCode(Normal::FAIL,'参数未配置');
        }
        
        $limit = (int)$config['SupplierPullCount'];
        $limit = $limit > 0 ? $limit : self::PULL_USER_COUNT;
        //所有供应商拉取完成间隔时间
        $SupplierPullWaitTime = $config['SupplierPullWaitTime'];
        $SupplierPullWaitTime = $SupplierPullWaitTime > 0 ? $SupplierPullWaitTime : self::TIME_WAIT;
        
        $page = 1;
        while (true)
        {
            $SupplierList = $this->getWaitPullSupplierWareHouseList($page,$limit,$SupplierID);
            if( empty($SupplierList) ) 
            {
                break;
            }
            
            //开始协助并发拉取
            $pull_supplier_count = count($SupplierList);
            $wg = new WaitGroup();
            $wg->add($pull_supplier_count);
            foreach ($SupplierList as $supplier)
            {
                co(function () use ($supplier,$SupplierPullWaitTime,$wg) 
                {
                    $this->getEverySupplierGoodsLists($supplier,$SupplierPullWaitTime);
                    $wg->done();
                });
            }
            $wg->wait();
            
            $page++;
        }
    }
    
    /**
     * 处理每一个供应商的商品数据
     * @param array $supplier
     */
    public function getEverySupplierGoodsLists($supplier = [],$SupplierPullWaitTime)
    {
        //获取供应商参数
        $company_code   = $supplier['ThirdPartyPlatformCode'] ? $supplier['ThirdPartyPlatformCode'] : $supplier['SupplierCode'];
        $SupplierID     = $supplier['SupplierID'];
        $newPage        = 1;
        $hasTotalWrite  = 0;
        
        //初始化拉取状态
        $updateArr = [
            'CommodityGetTaskStatus'    => 0,
            'CommodityGetFinishCount'   => 0,
            'CommodityGetFinishUseTime' => 0.000,
        ];
        SupplierCompany::instance()->where('ID',$SupplierID)->update($updateArr);
        
        $CommodityGetStartTime = microtime(true);
        
        $JsonBody = @json_decode($supplier['JsonBody'],true);
        $is_need_notify_last_page = 1;
        while ( true )
        {
            $queryBody = [
                'supplierID'    => $SupplierID,
                'page'          => $newPage,
                'apiBody'       => $JsonBody,
                'limit'         => $JsonBody['limit'] ?? 100,//默认100条
            ];
            
            try 
            {
                $resQuery = SupplierVendorApiGoodsService::instance()->queryGoodsList($queryBody);
                echo $resQuery['data']['has_next'].PHP_EOL;
                if( $resQuery['code'] != Normal::SUC )
                {
                    SupplierCompany::instance()->where('ID',$SupplierID)->update(['CommodityGetContents'=>$resQuery['msg']]);
                    if( $resQuery['data']['has_next'] ) 
                    {
                        $is_need_notify_last_page = 0;
                        $newPage++;
                        continue;
                    }
                    
                    break;
                }
                
                SupplierCompany::instance()->where('ID',$SupplierID)->update(['CommodityGetContents'=>$resQuery['msg']]);
            }
            catch ( \Throwable $t ) 
            {
                echo "当前{$company_code}--{$SupplierID}，第{$newPage}页\r\n";
                echo $t->getMessage().$t->getFile().$t->getLine().PHP_EOL;
                SupplierCompany::instance()->where('ID',$SupplierID)->update(['CommodityGetContents'=>$t->getMessage()]);
                break;
            }
            
            echo "当前{$company_code}--{$SupplierID}，第{$newPage}页\r\n";
            
            //更新总数量
            if( $hasTotalWrite != 1 ) 
            {
                $total = $resQuery['data']['total'];
                SupplierCompany::instance()->where('ID',$SupplierID)->update(['CommodityGetTotal'=>$total]);
            }
            
            //如果获取数据发送至消息队列，发送批量
            if( !empty($resQuery['data']['list']) && $resQuery['data']['has_next'] == true )
            {
                //动态更新已拉取的数据
                $pull_count = count($resQuery['data']['list']);
                $updateArr = [
                    'CommodityGetTaskStatus'    => 1,
                    'CommodityGetFinishCount'   => Db::raw("CommodityGetFinishCount+{$pull_count}")
                ];
                SupplierCompany::instance()->where('ID',$SupplierID)->update($updateArr);
                //标识是否有总数量更新
                $hasTotalWrite = 1;
                $MessageBody  = [
                    'SupplierID'        => $SupplierID,
                    'SupplierCode'      => $company_code,
                    'IsFirstPage'       => $newPage == 1 ? 1 : 0,
                    'DataPage'          => $newPage,
                    'IsLastPage'        => 0,
                    'GoodsItemList'     => $resQuery['data']['list']
                ];
                $is_need_notify_last_page = 1;
            }
            else
            {
                $IsLastPage = 1;
                $MessageBody  = [
                    'SupplierID'        => $SupplierID,
                    'SupplierCode'      => $company_code,
                    'IsFirstPage'       => 0,
                    'DataPage'          => $newPage,
                    'IsLastPage'        => $is_need_notify_last_page == 1 ?  1 : 0,
                    'GoodsItemList'     => []
                ];
                
                //预计下次拉取商品时间
                $CommodityGetNetPeriod = date("Y-m-d H:i:s",time() + $SupplierPullWaitTime * 60 );
                //拉取完成
                $updateArr = [
                    'CommodityGetTaskStatus'    => 2,
                    'CommodityGetContents'      => '拉取完成',
                    'CommodityGetFinishUseTime' => bcsub(microtime(true),$CommodityGetStartTime,3),
                    'CommodityGetNetPeriod'     => $CommodityGetNetPeriod
                ];
                SupplierCompany::instance()->where('ID',$SupplierID)->update($updateArr);
            }
        
            $this->publish($MessageBody);
            
            //最后一页
            if( isset($IsLastPage) ) 
            {
                break;
            }
            
            $newPage++;
        }
    }
    
    /**
     * 消息发布到MQ
     * @param array $body
     */
    public function publish( $MessageBody = [] )
    {
        $publishMsg = json_encode($MessageBody,JSON_UNESCAPED_UNICODE);
        try {
            $handlerMessage = new SupplierCommodityPullProducer($MessageBody);
            $producer = container()->get(Producer::class);
            $res = $producer->produce($handlerMessage,true);
            if( !$res ) {
                sleep(3);
                echo "publish failed \r\n";
                return $this->publish($MessageBody);
                return Normal::returnCode(Normal::FAIL,"发布失败");
            }
        } catch ( \Throwable $e ) {
            sleep(3);
            echo "publish failed {$e->getMessage()} \r\n";
            return $this->publish($MessageBody);
        }
    }
    
    /**
     * 获取需要拉取商品数据的接口供应商
     * @param int $page
     * @param int $limit
     * @param number $SupplierID
     */
    public function getWaitPullSupplierWareHouseList( int $page , int $limit, $SupplierID = 0 ) 
    {
        $table = WarehouseInfo::instance()->getTable();
        $query = WarehouseInfo::instance()
        ->from($table." as w")
        ->select([
            Db::raw("CONCAT(w.SupplierID,'') as SupplierID"),
            's.SupplierCode',
            's.SupplierName',
            's.OrderDockMode',
            's.IsTryUpdateCommodityArchives',
            's.ThirdPartyPlatformCode',
            's.SyncMode',
            Db::raw("CONCAT(w.ID,'') as WarehouseID"),
            'w.WarehouseName',
            'w.JsonBody',
            'w.CanModifyStock',
        ])
        ->join('suppliercompany_s2b as s',"s.ID","=","w.SupplierID");
        
        if( $SupplierID > 0 ) {
            $query->where('w.SupplierID',$SupplierID);
        }
        
        $pageInfo = $query
        ->where('s.CompanyStatus',1)//供应商启用状态
        ->where('s.OrderDockMode',0)//供应商接口仓
        ->where('s.CommodityGetMode',1)//供应商商品接口主动拉取模式
        ->where('w.WarehouseType',1)//接口模式
        ->where('w.ExamineState',1)//审核通过
        ->where('CommodityGetNetPeriod','<=',date('Y-m-d H:i:s'))
        ->orderBy('w.ID','ASC')
        ->paginate($limit, ['*'], 'page', $page,true)
        ->toArray();
        
        return $pageInfo['data'];
    }
    
    /**
     * 根据供应商编码寻找供应商参数
     * @param string $SupplierCode
     */
    public function getWaitPullSupplierWareHouseInfoBySupplierCode( string $SupplierCode )
    {
        $table = WarehouseInfo::instance()->getTable();
        $query = WarehouseInfo::instance()
        ->from($table." as w")
        ->select([
            Db::raw("CONCAT(w.SupplierID,'') as SupplierID"),
            's.SupplierCode',
            's.SupplierName',
            's.OrderDockMode',
            's.IsTryUpdateCommodityArchives',
            's.ThirdPartyPlatformCode',
            's.SyncMode',
            Db::raw("CONCAT(w.ID,'') as WarehouseID"),
            'w.WarehouseName',
            'w.JsonBody',
            'w.CanModifyStock',
        ])
        ->join('suppliercompany_s2b as s',"s.ID","=","w.SupplierID")
        ->where('s.SupplierCode',$SupplierCode)
        ->whereIn('s.CommodityGetTaskStatus',[0,2,3])
        ->where('s.CompanyStatus',1)//供应商启用状态
        ->where('s.OrderDockMode',0)//供应商接口仓
        ->where('s.CommodityGetMode',3)//供应商商品接口主动拉取模式
        ->where('w.WarehouseType',1)//接口模式
        ->where('w.ExamineState',1)
        ->where('CommodityGetNetPeriod','<=',date('Y-m-d H:i:s'))    
        ->first();
        
        $wareHouseParam = $query ? $query->toArray() : [];
        if( empty($wareHouseParam) ) {
            return Normal::returnCode(Normal::FAIL,"未找到满足条件的供应商数据");
        }
        
        //获取行云的所有分类
        $JsonBody = @json_decode($wareHouseParam['JsonBody'],true);
        if( empty($JsonBody) ) {
            return Normal::returnCode(Normal::FAIL,"供应商仓库参数未配置");
        }
        $wareHouseParam['JsonBody'] = $JsonBody;
        return Normal::returnCode(Normal::SUC,"获取成功",$wareHouseParam);
    }

    /**
     * 根据供应商编码寻找供应商参数
     * @param string $SupplierCode
     */
    public function getWaitPullSupplierWareHouseInfosByThirdPartyPlatformCode( string $ThirdPartyPlatformCode ,$SupplierID=0)
    {
        $table = WarehouseInfo::instance()->getTable();
        $where=[];
        if ($SupplierID>0){
            $where['w.SupplierID']=$SupplierID;
        }
        $wareHouseParam = WarehouseInfo::instance()
        ->from($table." as w")
        ->select([
            Db::raw("CONCAT(w.SupplierID,'') as SupplierID"),
            's.SupplierCode',
            's.SupplierName',
            's.OrderDockMode',
            's.IsTryUpdateCommodityArchives',
            's.ThirdPartyPlatformCode',
            's.SyncMode',
            Db::raw("CONCAT(w.ID,'') as WarehouseID"),
            'w.WarehouseName',
            'w.JsonBody',
            'w.CanModifyStock',
        ])
        ->join('suppliercompany_s2b as s',"s.ID","=","w.SupplierID")
        ->where('s.ThirdPartyPlatformCode',$ThirdPartyPlatformCode)
        ->whereIn('s.CommodityGetTaskStatus',[0,2,3])
        ->where('s.CompanyStatus',1)//供应商启用状态
        ->where('s.OrderDockMode',0)//供应商接口仓
        ->where('s.CommodityGetMode',3)//供应商商品接口主动拉取模式
        ->where('w.WarehouseType',1)//接口模式
        ->where('w.ExamineState',1)
        ->where('CommodityGetNetPeriod','<=',date('Y-m-d H:i:s'))
        ->where($where)
        ->get()
        ->toArray();
        
        if( empty($wareHouseParam) ) {
            return Normal::returnCode(Normal::FAIL,"未找到满足条件的供应商数据");
        }
        
        foreach ($wareHouseParam as $k=>&$v) 
        {
            //获取行云的所有分类
            $JsonBody = @json_decode($v['JsonBody'],true);
            if( empty($JsonBody) ) {
                unset($wareHouseParam[$k]);
            }
            $v['JsonBody'] = $JsonBody;
        }
        return Normal::returnCode(Normal::SUC,"获取成功",$wareHouseParam);
    }
}